package com.intct.util;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.kafka.clients.producer.ProducerConfig;

/**
 * @author gufg
 * @since 2025-08-19 09:51
 */
public class KafkaUtil {
    /**
     * 获取Kafka Sink
     * @param parameterTool
     * @param topic
     * @return
     */
    public static KafkaSink<String> getKafkaSink( ParameterTool parameterTool, String topic) {
        return KafkaSink.<String>builder()
                .setBootstrapServers(parameterTool.get("kafka-bootstrap-servers", "cdh-noe:9092"))
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.<String>builder()
                                .setTopic(topic)
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()

                )
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                .setTransactionalIdPrefix(topic)
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "90000")
                .build();
    }

    /**
     * 获取Kafka Source
     * @param parameterTool
     * @param topic
     * @param groupId
     * @return
     */
    public static KafkaSource<String> getKafkaSource(ParameterTool parameterTool, String topic, String groupId) {
        return KafkaSource.<String>builder()
                .setBootstrapServers(parameterTool.get("kafka-bootstrap-servers", "cdh-node:9092"))
                .setTopics(topic)
                .setGroupId(groupId)
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setStartingOffsets(OffsetsInitializer.earliest())
                .build();
    }
}
